package day2

import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}

object PairRDD {
  def main(args: Array[String]): Unit = {
    System.setProperty("hadoop.home.dir","D:\\hadoop")
    val sc = new SparkContext(new SparkConf().setAppName("Basic").setMaster("local"))
    Logger.getLogger("org").setLevel(Level.OFF)

    //PariRDD:键值对RDD，在Spark中，只能是一个二元组才能转换为PairRDD
    //*ByKey ByValue => 都是属于PariRDD独有的，*By => 表示共有

    val pair = sc.parallelize(List((1,2),(1,2),(1,3),(1,4),(1,5),(1,6),(2,1),(2,2),(2,3),(2,4),(2,5),(2,6)))

    val pair_int = sc.parallelize(List(1,2,3,4,5)).map(i =>(i,1)).foreach(println)

    val pair_string = sc.parallelize(List("yzy hello","spark hello","scala hello"))map(i =>{
      val list = i.split(" ")
      (list(0),i)
    })


//    val test  =sc.parallelize(List("yzy hello","spark hello","scala hello"))map(i =>{
//      val list = i.split(" ")
//      (list(0),i)
//    })

    //转换算子
    val pair_mapValues = pair.mapValues(println(_))

    //reduceByKey():接受一个函数.对于具有相同键的值进行操作
    //reduceByKey()算子返回一个PairRDD，输出的是一个相同键的值合并完成之后的值(k1,vs1),(k2,vs2)



    val pair_StringInt = sc.parallelize(List(('a',1),('b',2),('c',3),('d',4),('b',1),('c',2)))

    val pari_StringInt_reduceByKey = pair_StringInt.reduceByKey(
      (x,y) => x+y
    ).foreach(println)

    //countByValue 算子 基于相同的key,做值的合并
    //将一个标准的RDD,相同的值当做key，value输出为key的个数和
    pair_string.countByValue().foreach(println)
    //    val pair = sc.parallelize(List((1,2),(1,2),(1,3),(1,4),(1,5),(1,6),(2,1),(2,2),(2,3),(2,4),(2,5),(2,6)))
    pair.combineByKey(
      v => (v,1),
      (k : (Int,Int),v : Int) => (k._1+v,k._2+1),
      (k :(Int,Int),v :(Int,Int)) => (k._1+v._1,v._2+k._2)
    )
//      .foreach(println)
//      .mapValues(i =>println(i._1 / i._2)).count()
    //强制按照key排序
    pair_StringInt.sortByKey(ascending = false).foreach(println)
    //自定义排序
    pair_StringInt.sortBy(i => i).foreach(println)

    //groupByKey 算子 相同的key进行合并,值进行合并 (a,1)(a,2) => (a,Seq(1,2))

    //pair_StringInt.groupByKey().foreach(println)
    val pair_join = sc.parallelize(List(('a',5),('b',3),('c',2),('d',4),('b',7),('c',2)))
    //    val pair_StringInt = sc.parallelize(List(('a',1),('b',2),('c',3),('d',4),('b',1),('c',2)))
    //实现RDD合并
    //全连接
    pair_StringInt.join(pair_join).foreach(println)
    println("================================")
    //左连接
    pair_StringInt.leftOuterJoin(pair_join).foreach(println)
    println("===============================")
    //右连接
    pair_StringInt.rightOuterJoin(pair_join).foreach(println)
  }
}
